package mylab.io.store.socket.udp.multicast;

import com.google.common.base.Preconditions;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.MultiThreadIoEventLoopGroup;
import io.netty.channel.nio.NioIoHandler;
import io.netty.channel.socket.nio.NioDatagramChannel;
import mylab.io.base.model.Msg;
import mylab.io.base.util.DasUtil;
import mylab.io.core.AbstractClientDas;
import mylab.io.core.context.DefaultCC;
import mylab.io.store.socket.property.UdpMulticastProperties;

public class UdpMulticastDas<M extends Msg>
        extends AbstractClientDas<UdpMulticastProperties, M, UdpMulticastCodec<M>, ByteBuf> {

    public UdpMulticastDas(DefaultCC<UdpMulticastProperties, M> context, UdpMulticastCodec<M> codec) {
        super(context, codec);
    }

    @Override
    protected ChannelFuture actualStart() throws Exception {
        Preconditions.checkState(DasUtil.isValidEndpoint(properties.getGroup()), "invalid endpoint");

        masterLoop = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());

        bootstrap = new Bootstrap();
        bootstrap.group(masterLoop)
                .channel(NioDatagramChannel.class)
                .handler(codec.getChannelInitializer())
                .localAddress(properties.getGroup().getPort())
                .option(ChannelOption.IP_MULTICAST_IF, properties.getNetworkInterface())
                .option(ChannelOption.IP_MULTICAST_LOOP_DISABLED, properties.isLoopbackModeDisabled())
                .option(ChannelOption.SO_REUSEADDR, properties.isReuseAddress())
                .option(ChannelOption.IP_MULTICAST_TTL, properties.getTtl());

        ChannelFuture future = bootstrap.bind(properties.getGroup().getPort()).sync();
        NioDatagramChannel datagramChannel = (NioDatagramChannel) future.channel();
        return datagramChannel.joinGroup(properties.getGroup().getAddress(), properties.getNetworkInterface());
    }
}
